package com.atguigu.bigdata.spark.core.seq

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

import scala.collection.mutable.ArrayOps

object Spark_req_hotTop10_1 {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("operator")
    val sc = new SparkContext(sparkConf)

    //TODO TOP10热门商品

    //1.获取数据
    val dataRDD: RDD[String] = sc.textFile("data/user_visit_action.txt")


    //2.统计不同维度数据[详情点击，下单点击，支付点击 ]的数量

    //2.1品类 点击数量
    //过滤详情点击的数据
    val clickData: RDD[String] = dataRDD.filter(
      line => {
        val datas: Array[String] = line.split("_")
        datas(6) != "-1"
      }
    )
    val clickCount: RDD[(String, Int)] = clickData.map(
      line => {
        val datas: Array[String] = line.split("_")
        (datas(6), 1)
      }
    ).reduceByKey(_+_)


    //2.2品类 下单点击
    val orderData: RDD[String] = dataRDD.filter(
      line => {
        val datas: Array[String] = line.split("_")
        datas(8) != "null"
      }
    )
    val orderCount: RDD[(String, Int)] = orderData.flatMap(
      line => {
        val datas: Array[String] = line.split("_")
        val ids = datas(8).split(",")
        ids.map((_,1))

      }
    ).reduceByKey(_ + _)




    //2.3品类 支付点击
    val payData: RDD[String] = dataRDD.filter(
      line => {
        val datas: Array[String] = line.split("_")
        datas(10) != "null"
      }
    )
    val payCount: RDD[(String, Int)] = payData.flatMap(
      line => {
        val datas: ArrayOps.ofRef[String] = line.split("_")
        val ids: ArrayOps.ofRef[String] = datas(10).split(",")
        ids.map((_, 1))
      }
    ).reduceByKey(_+_)
    payCount


    //3按照点击，下单，支付，各个排序。
    /*    val sortRDD: RDD[(String, Int)] = clickCount.sortBy(_._2,false)*/
    /**2**********************************************************************/
      //解决了因为某个排序方式的数据为0而进行的join,rdd算子为0的状态。
    val cogroup: RDD[(String, (Iterable[Int], Iterable[Int], Iterable[Int]))] = clickCount.cogroup(orderCount,payCount)

    cogroup.mapValues{
      case (click,order,pay)=>{
       var clickCnt=0
       var orderCnt=0
       var payCnt=0

        val iterator1: Iterator[Int] = click.iterator
        if(iterator1.hasNext){
          clickCnt = iterator1.next()
        }

        val iterator2: Iterator[Int] = order.iterator
        if(iterator2.hasNext){
         orderCnt = iterator2.next()
        }

        val iterator3: Iterator[Int] = pay.iterator
        if(iterator3.hasNext){
           payCnt = iterator3.next()
        }

        (clickCnt,orderCnt,payCnt)
      }
    }.sortBy(_._2,false).take(10).foreach(println)



    sc.stop()
  }
}
